package serv_impl

import (
	"context"
	"io"
	"time"

	"atomgit.com/openlinksaas/proto-gen-go.git/k8s_proxy_api"
	"github.com/google/uuid"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	schema2 "k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
)

type SizeQueue struct {
	resizeChan chan remotecommand.TerminalSize
}

func newSizeQueue(width, height uint16) *SizeQueue {
	ret := &SizeQueue{
		resizeChan: make(chan remotecommand.TerminalSize, 100),
	}
	ret.resizeChan <- remotecommand.TerminalSize{
		Width:  width,
		Height: height,
	}
	return ret
}

func (s *SizeQueue) ChangeSize(width, height uint16) {
	s.resizeChan <- remotecommand.TerminalSize{
		Width:  width,
		Height: height,
	}
}

func (s *SizeQueue) Next() *remotecommand.TerminalSize {
	size, ok := <-s.resizeChan
	if !ok {
		return nil
	}
	return &size
}

func (s *SizeQueue) Close() {
	close(s.resizeChan)
}

type TermItem struct {
	TermId       string
	Cancel       context.CancelFunc
	StdinReader  *io.PipeReader
	StdinWriter  *io.PipeWriter
	OutChan      chan []byte
	SizeQueue    *SizeQueue
	LastReadTime int64
}

type StdoutWriter struct {
	dataChan chan []byte
}

func (w *StdoutWriter) Write(p []byte) (n int, err error) {
	defer func() {
		recover()
	}()

	w.dataChan <- p
	return len(p), nil
}

func (apiImpl *K8SProxyApiImpl) OpenTerm(ctx context.Context, req *k8s_proxy_api.OpenTermRequest) (*k8s_proxy_api.OpenTermResponse, error) {
	tokenInfo, err := getTokenInfo(apiImpl.linkSaasAddr, req.Token)
	if err != nil {
		return nil, err
	}
	k8sClientSet, err := getK8sClientSet(apiImpl.kubeConfigFile)
	if err != nil {
		return nil, err
	}
	{
		if req.Namespace == "" {
			return &k8s_proxy_api.OpenTermResponse{
				Code:   k8s_proxy_api.OpenTermResponse_CODE_NO_PERMISSION,
				ErrMsg: "必须指定命名空间",
			}, nil
		}
		client := k8sClientSet.CoreV1().Namespaces()
		res, err := client.Get(ctx, req.Namespace, metav1.GetOptions{})
		if err != nil {
			return nil, err
		}
		if res.Labels[PROJECT_LABEL] != tokenInfo.ProjectId {
			return &k8s_proxy_api.OpenTermResponse{
				Code:   k8s_proxy_api.OpenTermResponse_CODE_NO_PERMISSION,
				ErrMsg: "命名空间没有关联项目",
			}, nil
		}
	}
	if !tokenInfo.Admin {
		userPerm := apiImpl.getResourceUserPerm(ctx, req.Namespace, req.ResourceType, req.ResourceName, tokenInfo.MemberUserId)
		if !userPerm.UpdateScale {
			return &k8s_proxy_api.OpenTermResponse{
				Code:   k8s_proxy_api.OpenTermResponse_CODE_NO_PERMISSION,
				ErrMsg: "没有权限",
			}, nil
		}
	}

	cfg, err := clientcmd.BuildConfigFromFlags("", apiImpl.kubeConfigFile)
	if err != nil {
		return nil, err
	}
	cfg.APIPath = "/api"
	cfg.NegotiatedSerializer = scheme.Codecs.WithoutConversion()
	cfg.GroupVersion = &schema2.GroupVersion{Version: "v1"}
	cfg.UserAgent = rest.DefaultKubernetesUserAgent()

	client, err := rest.RESTClientFor(cfg)
	if err != nil {
		return nil, err
	}
	termReq := client.Post().Resource("pods").Name(req.PodName).Namespace(req.Namespace).SubResource("exec")
	termReq.VersionedParams(&corev1.PodExecOptions{
		Container: req.ContainerName,
		Stdin:     true,
		Stdout:    true,
		Stderr:    true,
		TTY:       true,
		Command:   []string{"/bin/sh"},
	}, scheme.ParameterCodec)

	exec, err := remotecommand.NewSPDYExecutor(cfg, "POST", termReq.URL())
	if err != nil {
		return nil, err
	}

	termCtx, cancel := context.WithCancel(context.Background())
	stdinReader, stdinWriter := io.Pipe()
	outChan := make(chan []byte, 100)
	sizeQueue := newSizeQueue(uint16(req.TermWidth), uint16(req.TermHeight))

	termId := uuid.NewString()
	apiImpl.termItemMap.Store(termId, &TermItem{
		TermId:       termId,
		Cancel:       cancel,
		StdinReader:  stdinReader,
		StdinWriter:  stdinWriter,
		OutChan:      outChan,
		SizeQueue:    sizeQueue,
		LastReadTime: time.Now().Unix(),
	})

	go apiImpl.termStream(termCtx, exec, stdinReader, outChan, sizeQueue)

	return &k8s_proxy_api.OpenTermResponse{
		Code:   k8s_proxy_api.OpenTermResponse_CODE_OK,
		TermId: termId,
	}, nil
}

func (apiImpl *K8SProxyApiImpl) termStream(ctx context.Context, exec remotecommand.Executor, stdinReader io.Reader, outChan chan []byte, sizeQueue *SizeQueue) {
	defer func() {
		recover()
	}()

	exec.StreamWithContext(ctx, remotecommand.StreamOptions{
		Stdin:             stdinReader,
		Stdout:            &StdoutWriter{dataChan: outChan},
		Stderr:            &StdoutWriter{dataChan: outChan},
		Tty:               true,
		TerminalSizeQueue: sizeQueue,
	})
}

func (apiImpl *K8SProxyApiImpl) WriteTerm(ctx context.Context, req *k8s_proxy_api.WriteTermRequest) (*k8s_proxy_api.WriteTermResponse, error) {
	defer func() {
		recover()
	}()

	termItemValue, ok := apiImpl.termItemMap.Load(req.TermId)
	if !ok {
		return &k8s_proxy_api.WriteTermResponse{
			Code:   k8s_proxy_api.WriteTermResponse_CODE_NO_TERM,
			ErrMsg: "终端句柄不存在",
		}, nil
	}
	termItem := termItemValue.(*TermItem)
	termItem.StdinWriter.Write(req.Data)
	return &k8s_proxy_api.WriteTermResponse{
		Code: k8s_proxy_api.WriteTermResponse_CODE_OK,
	}, nil
}

func (apiImpl *K8SProxyApiImpl) ReadTerm(ctx context.Context, req *k8s_proxy_api.ReadTermRequest) (*k8s_proxy_api.ReadTermResponse, error) {
	defer func() {
		recover()
	}()
	termItemValue, ok := apiImpl.termItemMap.Load(req.TermId)
	if !ok {
		return &k8s_proxy_api.ReadTermResponse{
			Code:   k8s_proxy_api.ReadTermResponse_CODE_NO_TERM,
			ErrMsg: "终端句柄不存在",
			Data:   []byte{},
		}, nil
	}
	termItem := termItemValue.(*TermItem)
	termItem.LastReadTime = time.Now().Unix()
	apiImpl.termItemMap.Store(termItem.TermId, termItem)

	select {
	case data := <-termItem.OutChan:
		return &k8s_proxy_api.ReadTermResponse{
			Code: k8s_proxy_api.ReadTermResponse_CODE_OK,
			Data: data,
		}, nil
	case <-time.After(30 * time.Second):
		return &k8s_proxy_api.ReadTermResponse{
			Code: k8s_proxy_api.ReadTermResponse_CODE_OK,
			Data: []byte{},
		}, nil
	}
}

func (apiImpl *K8SProxyApiImpl) SetTermSize(ctx context.Context, req *k8s_proxy_api.SetTermSizeRequest) (*k8s_proxy_api.SetTermSizeResponse, error) {
	defer func() {
		recover()
	}()

	termItemValue, ok := apiImpl.termItemMap.Load(req.TermId)
	if !ok {
		return &k8s_proxy_api.SetTermSizeResponse{
			Code:   k8s_proxy_api.SetTermSizeResponse_CODE_NO_TERM,
			ErrMsg: "终端句柄不存在",
		}, nil
	}
	termItem := termItemValue.(*TermItem)
	termItem.SizeQueue.ChangeSize(uint16(req.TermWidth), uint16(req.TermHeight))
	return &k8s_proxy_api.SetTermSizeResponse{
		Code: k8s_proxy_api.SetTermSizeResponse_CODE_OK,
	}, nil
}
